#include <iostream>
#include <fcntl.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <cstring>

#include "sunnet.h"

using namespace std;

// 单例
// 所以静态变量都需要在这里定义, c++ 特有的写法.
sunnet *sunnet::inst;
sunnet::sunnet()
{
    inst = this;
}

// 启动函数
void sunnet::start()
{
    cout << "Hello, Sunnet" << endl;

    start_worker();
    // 开启后的一些其他逻辑...
    // ...
    // 锁
    pthread_rwlock_init(&this->services_lock, NULL); // null, 表示使用默认属性
    pthread_spin_init(&this->global_queue_lock, PTHREAD_PROCESS_PRIVATE);
    pthread_mutex_init(&this->sleep_mutex_lock, NULL);
    pthread_cond_init(&this->sleep_cond, NULL);
    pthread_rwlock_init(&this->conns_lock, NULL);

    start_socket_worker();
}

void sunnet::wait()
{
    if (worker_threads[0])
    {
        worker_threads[0]->join();
    }
}

void sunnet::start_worker()
{
    for (int i = 0; i < worker_num; i++)
    {
        cout << "start worker id :" << i << endl;
        // 创建线程对象
        worker *w = new worker();
        w->id = i;
        w->each_num = 2 << i;
        // 创建线程
        thread *wt = new thread(*w);
        // 添加到数组
        workers.push_back(w);
        worker_threads.push_back(wt);
    }
}

void sunnet::start_socket_worker()
{
    // 这里为什么不使用 smart 指针呢？
    // 这是因为 this->soc_worker 和 this->soc_thread 与进程同生共死, 无需释放.
    this->soc_worker = new socket_worker();
    this->soc_worker->init();
    this->soc_thread = new thread(*this->soc_worker);
}

uint32_t sunnet::newservice(shared_ptr<string> type)
{
    // 标准库中的方法是否线程安全取决于是否修改了共享数据
    // 同理, 自定义方法也是如此
    // 以下 make_shared 方法本身是线程安全的
    shared_ptr<service> srv = make_shared<service>();
    srv->type = type;
    pthread_rwlock_wrlock(&this->services_lock);
    {
        srv->id = this->maxid;
        this->maxid++;
        this->services.emplace(srv->id, srv);
    }
    pthread_rwlock_unlock(&this->services_lock);
    // 根据自己的理解, 这里可能是线程不安全的(thread-unsafe), 不过, 按照 Actor 设计
    // 此时, srv 还没有加入全局队列, 故此, 别的线程拿不到它, 因为它是局部变量
    srv->on_init();
    return srv->id;
}

shared_ptr<service> sunnet::get_service(uint32_t id)
{
    shared_ptr<service> srv = nullptr;
    pthread_rwlock_rdlock(&this->services_lock);
    {
        unordered_map<uint32_t, shared_ptr<service>>::iterator it = services.find(id);
        if (it != services.end())
        {
            srv = it->second;
        }
    }
    pthread_rwlock_unlock(&this->services_lock);
    return srv;
}

void sunnet::killservice(uint32_t id)
{
    shared_ptr<service> srv = this->get_service(id);
    if (!srv)
    {
        return;
    }

    srv->on_exit();
    srv->is_existing = true;
    pthread_rwlock_wrlock(&this->services_lock);
    {
        services.erase(id);
    }
    pthread_rwlock_unlock(&this->services_lock);
}

shared_ptr<service> sunnet::pop_global_queue()
{
    shared_ptr<service> srv = nullptr;
    pthread_spin_lock(&this->global_queue_lock);
    {
        if (!this->global_queue.empty())
        {
            srv = this->global_queue.front();
            this->global_queue.pop();
            this->global_queue_len--;
        }
    }
    pthread_spin_unlock(&this->global_queue_lock);

    return srv;
}

void sunnet::push_global_queue(shared_ptr<service> srv)
{
    pthread_spin_lock(&this->global_queue_lock);
    {
        this->global_queue.push(srv);
        this->global_queue_len++;
    }
    pthread_spin_unlock(&this->global_queue_lock);
}

void sunnet::send(uint32_t id, shared_ptr<base_msg> msg)
{
    shared_ptr<service> to_service = this->get_service(id);
    if (!to_service)
    {
        cout << "send fail, " << "service " << id << "is not exist." << endl;
        return;
    }

    // 往目标服务的消息队列插入消息
    to_service->push_msg(msg);

    // 检查并放入全局队列
    bool has_push = false;
    pthread_spin_lock(&to_service->is_in_global_queue_lock);
    {
        if (!to_service->is_in_global_queue)
        {
            this->push_global_queue(to_service);
            to_service->is_in_global_queue = true;
            has_push = true;
        }
    }
    pthread_spin_unlock(&to_service->is_in_global_queue_lock);

    // 唤醒线程, 不放在临界区
    if (has_push)
    {
        this->check_and_weakup();
    }
}

// 仅测试使用
shared_ptr<base_msg> sunnet::make_msg(uint32_t source, char *buff, size_t size)
{
    auto msg = make_shared<service_msg>();
    msg->type = base_msg::TYPE::service;
    msg->source = source;
    // 基本类型的数据(char, int, ...)没有析构函数
    // delete 或者 delete[] 都可以销毁基本类型的数据组成的数组
    // shared_ptr 默认使用 delete.
    // 所以, 此处无需重写 smart pointer 的销毁方法
    msg->buff = shared_ptr<char>(buff);
    msg->size = size;

    return msg;
}

void sunnet::worker_wait()
{
    pthread_mutex_lock(&this->sleep_mutex_lock);
    this->sleep_count++;
    pthread_cond_wait(&this->sleep_cond, &this->sleep_mutex_lock); // 这里包含解锁, 唤醒, 加锁
    this->sleep_count--;
    pthread_mutex_unlock(&this->sleep_mutex_lock);
}

void sunnet::check_and_weakup()
{
    // 这里对 sleep_count 和 global_queue_len 的读取是不精确的
    // 但是问题不大, 因为锁的开销更大,
    // 所以这里, 宁愿出错也不加锁.
    int sleep_count = this->sleep_count;
    int global_queue_len = this->global_queue_len;

    if (sleep_count == 0)
        return;

    // 这里参考 skynet 是如何唤醒新的线程的, 是如何对数量进行判断的 ???
    int still_alive_worker_num = this->worker_num - sleep_count;
    if (still_alive_worker_num <= global_queue_len) // <= 操作符 表示 如果正在工作的线程数量大于需要处理的 Actor 数, 那么就认为足以应对
        pthread_cond_signal(&this->sleep_cond);     // 随机唤醒一个线程
}

/**
 * 试讨论方法 Listen 的线程安全性:
 * 它是给工作线程调用的, 业务层(Actor)会调用它以监听端口.
 *
 * 它本身未使用任何全局变量, 因此它是线程安全的.
 * 如果 epoll_ctl 和 epoll_wait 被多条线程调用, 按照书所说, 它们将由操作系统保证线程安全.
 * 故此, 该方法是线程安全的.
 * over.
 *
 * 注意, skynet 的方式不一样, 它使用管道作为网络线程与工作线程之间的通信, 工作线程并不直接调用监听方法,
 * 以保证线程安全.
 */
int sunnet::Listen(uint32_t port, uint32_t serviceid)
{
    // 1. 创建 socket
    int listenfd = socket(AF_INET, SOCK_STREAM, 0);
    if (listenfd <= 0)
    {
        cout << "listen error, listenfd <= 0" << endl;
        return -1;
    }

    // 2. 设置为非阻塞
    fcntl(listenfd, F_SETFL, O_NONBLOCK);
    // 3. bind
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    int r = bind(listenfd, (struct sockaddr *)&addr, sizeof(addr));
    if (r == -1)
    {
        cout << "listen error, bind fail " << strerror(errno) << endl;
        return -1;
    }
    // 4. listen
    r = listen(listenfd, 64);
    if (r < 0)
    {
        return -1;
    }
    // 5. 添加到管理结构
    this->add_conn(listenfd, serviceid, conn_type::listen);
    // 6. epoll 事件
    this->soc_worker->add_event(listenfd);

    return listenfd;
}

void sunnet::close_conn(uint32_t fd)
{
    // delete conn
    bool succ = this->remove_conn(fd);
    // close socket
    close(fd);
    // remove from epoll list
    if (succ)
    {
        this->soc_worker->remove_event(fd);
    }
}

int sunnet::add_conn(int fd, uint32_t serviceid, conn_type type)
{
    auto conninst = make_shared<conn>();
    conninst->fd = fd;
    conninst->serviceid = serviceid;
    conninst->type = type;
    pthread_rwlock_wrlock(&this->conns_lock);
    {
        this->conns.emplace(fd, conninst);
    }
    pthread_rwlock_unlock(&this->conns_lock);

    return fd;
}

shared_ptr<conn> sunnet::get_conn(int fd)
{
    shared_ptr<conn> conninst = nullptr;
    pthread_rwlock_rdlock(&this->conns_lock);
    {
        unordered_map<uint32_t, shared_ptr<conn>>::iterator it = this->conns.find(fd);
        if (it != this->conns.end())
        {
            conninst = it->second;
        }
    }
    pthread_rwlock_unlock(&this->conns_lock);

    return conninst;
}

bool sunnet::remove_conn(int fd)
{
    int result;
    pthread_rwlock_wrlock(&this->conns_lock);
    {
        result = this->conns.erase(fd);
    }
    pthread_rwlock_unlock(&this->conns_lock);

    return result == 1;
}

void sunnet::modify_event(int fd, bool epollOut)
{
    this->soc_worker->modify_event(fd, epollOut);
}

// 在 luaapi 中无法获取到服务对象, 但是需要调用服务的接口 on_socket_close, 所以在这里处理
// void sunnet::on_socket_close(uint32_t id)
// {
//     shared_ptr<service> srv = get_service(id);
// }